package com.wudl.flink;


import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @ClassName : FlinkCdc
 * @Description :
 * @Author :wudl
 * @Date: 2021-03-13 09:17
 */

public class FlinkCdc {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 2000L));

        // 设置后端状态
        env.setStateBackend(new FsStateBackend("hdfs://master01.pxx.com:8020/flinkCDC"));
        System.setProperty("HADOOP_USER_NAME", "hdfs");

        Properties prop = new Properties();
        prop.setProperty("scan.startup.mode", "initial");


        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("192.168.1.153")
                .port(3306)
                .username("root")
                .password("slave012021@123ABM")
                .databaseList("test")
                .tableList("test.test02")
                .deserializer(new StringDebeziumDeserializationSchema())
                .debeziumProperties(prop)
                .build();

        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        streamSource.print();
        env.execute("FlinkCdc");


    }
}
